/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.wsinfra.wsrm.impl;

import java.util.Arrays;
import java.util.Vector;

import cgl.narada.wsinfra.schemas.wsrm.SequenceAcknowledgementDocument;
import cgl.narada.wsinfra.wsrm.WsrmAckOperations;
import cgl.narada.wsinfra.wsrm.WsrmElementCreation;
import cgl.narada.wsinfra.wsrm.WsrmProcessingFactory;
import cgl.narada.wsinfra.wsrm.storage.WsrmProtocolStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageException;


/** A utility class to deal with processing acknowledgements.

    @author Shrideep Pallickara
    $Date: 2005/07/29 22:41:30 $
    $Revision: 1.5 $
*/


public class WsrmAckOperationsImpl implements WsrmAckOperations {
  
  private static final WsrmAckOperations instance = new WsrmAckOperationsImpl();
  private WsrmElementCreation wsrmElementCreation;

  private String moduleName = "WsrmAckOperations: ";

  private WsrmAckOperationsImpl() {

  }

  public static WsrmAckOperations getInstance() {
    return instance;
  }

  private void initializeWsrmElementCreation() {
    wsrmElementCreation = WsrmProcessingFactory.getWsrmElementCreation();
  }

  /** Construct acknowledgement ranges based on the specified acknowledgements.
   */
  public SequenceAcknowledgementDocument.SequenceAcknowledgement.AcknowledgementRange[]
  getAcknowledgementRanges(long[] acknowledgements) {
    
    Vector seperateAcks = new Vector();
    Vector ackRanges = new Vector();
    processAcks(acknowledgements, seperateAcks, ackRanges);
    
    int numOfSeperateAcks = seperateAcks.size();
    if (numOfSeperateAcks > 0) {
      long[] sepAcks = new long[numOfSeperateAcks];
      for (int i=0; i < numOfSeperateAcks; i++) {
	sepAcks[i] = ((Long)seperateAcks.elementAt(i)).longValue();
	createAndAddAcknowledgementRange(sepAcks[i], sepAcks[i], ackRanges);
      } 
    }
    

    return (SequenceAcknowledgementDocument.SequenceAcknowledgement.AcknowledgementRange[]) ackRanges.toArray(new SequenceAcknowledgementDocument.SequenceAcknowledgement.AcknowledgementRange[0]);

  }
  

  /** This method processes a set of acknowledgements, and splits them up into
      two distinct set of Vectors. The first Vector contains seperate 
      acknowledgements, while the second one contains a set of acknowledgement
      ranges. */
  private void processAcks(long[] acks, 
			   Vector seperateAcks, Vector ackRanges) {
    if (acks == null) {
      return;
    }
    
    /** Deal with the case where there is only one message number to process*/
    if (acks.length == 1) {
      seperateAcks.addElement(new Long(acks[0]));
      return;
    }

    /** First sort the acks, since the algo will work on the sorted array */
    Arrays.sort(acks);
    
    long lowerRange, upperRange;
    for (int i=0; i < acks.length -1; ) {
      lowerRange = acks[i];
      
      /** If successive elements are seperated by a value of 1 they can be
	  part of a range */
      while (acks[i] + 1 == acks[i+1]) {
	i++;
	if (i >= acks.length -1 ) {
	  break;
	}
      }
      upperRange = acks[i];
      
      if (lowerRange == upperRange) {
	/** This implies that this value should be a standalone value */
	seperateAcks.addElement(new Long(acks[i]));
	//System.out.print(" " + acks[i] + " ");
	i++;
	
	if (i == acks.length -1) {
	  seperateAcks.addElement(new Long(acks[i]));
	  //System.out.print(" " + acks[i] + " ");
	}
	
	if (i >= acks.length -1 ) {
	  break;
	}
	
      } else {
	//System.out.print(" R (" + lowerRange + ")<->("  + upperRange + ")");
	
	createAndAddAcknowledgementRange(upperRange, lowerRange, ackRanges);
	i++;
	if (i >= acks.length -1 ) {
	  break;
	}
      }
      
    }/* end for */    
  }/*end method */


  /** Create and add the acknowledgement range based on the parameters and add
      to the specified vector */  
  private void 
  createAndAddAcknowledgementRange(long upperRange, long lowerRange,
				   Vector ackRangeVector) {
    if (wsrmElementCreation == null) {
      System.out.println(moduleName + "WsrmCreation is NULL");
      initializeWsrmElementCreation();
    }
    SequenceAcknowledgementDocument.SequenceAcknowledgement.AcknowledgementRange acknowledgementRange =
      wsrmElementCreation.newAcknowledgementRange(upperRange, lowerRange);
    
    ackRangeVector.addElement(acknowledgementRange);
  }





  /** Retrieve the list of positive acknowledgements from the specified
      sequenceAcknowledgement document */
  public long[] 
  getAcknowledgements(SequenceAcknowledgementDocument seqAckDocument) {
    SequenceAcknowledgementDocument.SequenceAcknowledgement seqAcknowledgement 
      = seqAckDocument.getSequenceAcknowledgement();

    SequenceAcknowledgementDocument.SequenceAcknowledgement.AcknowledgementRange[] acknowledgementRanges = 
      seqAcknowledgement.getAcknowledgementRangeArray();
    
    Vector seperateAcknowledgements = new Vector();
    long[] seperateAcks = null;

    int numOfRanges = acknowledgementRanges.length;
    for (int i=0; i < numOfRanges; i++) {
      processAcknowledgementRange(acknowledgementRanges[i], 
				  seperateAcknowledgements);
    }

    int numOfSeperateAcks = seperateAcknowledgements.size();
    
    if (numOfSeperateAcks > 0) {
      seperateAcks = new long[numOfSeperateAcks];
      for (int i=0; i < numOfSeperateAcks; i++) {
	seperateAcks[i] = 
	  ((Long)seperateAcknowledgements.elementAt(i)).longValue();
      } 
    }
    
    if (seperateAcks != null) {
      Arrays.sort(seperateAcks);
    }

    return seperateAcks;
  }
  
  
  /** Process an acknowledgement range by trying to add the elements in the
      sepecified range into the seperateIntoVector. Also check to see if the
      element already exists in the vector, if so do not add */
  private void 
  processAcknowledgementRange(SequenceAcknowledgementDocument.SequenceAcknowledgement.AcknowledgementRange acknowledgementRange, 
			      Vector seperateIntoVector) {
    long lower = acknowledgementRange.getLower().longValue();
    long upper = acknowledgementRange.getUpper().longValue();
    
    for (long i=lower; i <= upper; i++) {
      Long toAdd = new Long(i);
      if (!seperateIntoVector.contains(toAdd)) {
	seperateIntoVector.addElement(toAdd);
      }     
    } /*end for */
  }



  
  /** Retrive the set of message numbers that need to be acknowledged for 
      a sequence. The set of message mnumbers returned is the union of two sets
      <br>(a)The message numbers that have already been acknowledged so far.
      <br>(b)The set of unacknowledged message numbers. If the 
      <i>ackRequested</i> variable is false, the message numbers included 
      with this set are those for which the ackInterval has expired. 
      <br>If the <i>ackRequested</i> variable if it is <i> true</i>, THEN the
      ackInterval constraint is ignored, and ALL message numbers that have not
      been acknowledged are included in the set.*/
  public long[] 
  getMessageNumbersToAcknowledge(String sequenceIdentifier, 
				 boolean ackRequested,
				 WsrmProtocolStorageOperations wsrmProtocolOps)
    throws WsrmStorageException {
    
    if (sequenceIdentifier == null || wsrmProtocolOps == null) {
      /** This will not happen, the code calling this ensures that this
	  will NOT happen */
      System.out.println(moduleName + "Specified sequence identifier = " +
			 sequenceIdentifier + ", wsrmProtocolOps = " +
			 wsrmProtocolOps);
      return null;
    }
    
    
    long[] messageNumbersToAck = null;
    
    long[] ackedMessageNumbers = 
      wsrmProtocolOps.getAcknowledgedMessageNumbers(sequenceIdentifier);
    
    long[] unackedMessageNumbers = 
      wsrmProtocolOps.getUnacknowledgedMessageNumbers(sequenceIdentifier, 
						      ackRequested);
    
    messageNumbersToAck = getCombinedArrayOfLong(ackedMessageNumbers,
						 unackedMessageNumbers);
    
    return messageNumbersToAck;    
  }

  

  /** Combines the two arrays to create a unique array without any
      duplicate values in the combined array. This will also eliminate any
      duplicates that might exist in the original arrays */
  private long[]
  getCombinedArrayOfLong(long[] first, long[] second) {
    if (first == null) {
      return second;
    }
    
    if (second == null) {
      return first;
    }
    
    Vector combinedVector = new Vector();
    for (int i=0; i < first.length; i++) {
      addToVector(combinedVector, first[i]);
    }
    
    for (int i=0; i < second.length; i++) {
      addToVector(combinedVector, second[i]);
    }
    
    if (combinedVector.size() == 0) {
      return null;
    }
    
    int numOfElements = combinedVector.size();
    long[] mergedArray = new long[numOfElements];
    for (int i=0; i < numOfElements; i++) {
      mergedArray[i] = ((Long)combinedVector.elementAt(i)).longValue();
    }
    
    return mergedArray;
  }
  
  /** Adds a message number to the vector */
  private static void addToVector(Vector vector, long value) {
    Long toAdd = new Long(value);
    if (vector.contains(toAdd)) {
      return;
    }
    vector.addElement(toAdd);
  }


}
